package com.zf.nospring.rabbitmq.dead_letter;

import com.rabbitmq.client.*;
import com.zf.nospring.rabbitmq.utils.RabbitUtils;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 消费正常队列
 */
public class NormalConsumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitUtils.getDefaultConnection();
        Channel channel = connection.createChannel();

        // 消费的逻辑
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                long deliveryTag = envelope.getDeliveryTag();
                Channel ch = getChannel();
                // 根据奇偶确认或者拒绝消息
                String msg = new String(body);
                if (deliveryTag % 2 == 0) {
                    // 手动确认消息
                    ch.basicAck(deliveryTag, false);
                    System.out.printf("正常消息：[%s] Envelope：%s, properties：%s\n", msg, envelope, properties);
                } else {
                    // 拒绝消息，并不再重新入队
                    ch.basicReject(deliveryTag, false);
                    System.out.printf("被拒绝消息：[%s]，将被路由到死信队列或者丢弃\n", msg);
                }
            }
        };

        // 消费正常队列，关闭自动确认
        channel.basicConsume("test_normal_queue", false, consumer);
    }
}
